﻿#include "../../../includes/multimedia/rtmp/crtmpserver.h"
#include "../../../includes/QsLog/QsLog.h"
#include "../../../includes/common/common.h"
#include "../../../includes/multimedia/rtmp/rtmp.h"
#include "../../../includes/multimedia/rtmp/rtmpchunk.h"
#include "../../../includes/multimedia/rtmp/crtmpclient.h"

#include <QDebug>
#include <QFileInfo>

/////////////////////////////////////////////////////////////////////////////////////////

cRtmpServer::cRtmpServer(QObject *parent)
    : QTcpServer(parent),
      avc_sequence_header_size_(0),
      max_chunk_size_(60000)
{
    QObject::connect(&m_RtmpServerHeartTimeOutTimer,
                     &QTimer::timeout,
                     this,
                     &cRtmpServer::handleRtmpServerHeartTimeOut);
    m_RtmpServerHeartTimeOutTimer.setInterval(1000);
}

cRtmpServer::~cRtmpServer()
{
    CloseServer();
}

/**
 * @brief cRtmpServer::OpenServer 打开一个指定端口的服务器
 * @param port 端口号
 */
bool cRtmpServer::OpenServer(int port)
{
    if(this->isListening() || port <= 0)
        return false;

    if (!this->listen(QHostAddress::Any, port))
    {
        showLog("cRtmpServer::OpenServer:"+
                QString::asprintf("%d",port)+
                " fail.");
        return false;
    }

    deleteAllClients();
    m_RtmpServerHeartTimeOutTimer.start();

    showLog("cRtmpServer::OpenServer:"+
            QString::asprintf("%d",port));
    return true;
}

void cRtmpServer::displayError(QAbstractSocket::SocketError socketError)
{
    showLog("cRtmpServer::displayError:" +
            QString::asprintf("%d",socketError));
}

/**
 * @brief CloseServer 关闭服务器
 */
void cRtmpServer::CloseServer(void)
{
    showLog(QString::fromLocal8Bit("服务器已经关闭."));

    m_RtmpServerHeartTimeOutTimer.stop();
    this->close();
    deleteAllClients();
}

void cRtmpServer::SetAvcSequenceHeader(QByteArray avcSequenceHeader, uint32_t avcSequenceHeaderSize)
{
    avc_sequence_header_ = avcSequenceHeader;
    avc_sequence_header_size_ = avcSequenceHeaderSize;
}

void cRtmpServer::incomingConnection(qintptr socketDescriptor) //多线程必须在此函数里捕获新连接
{
    cRtmpClient* tcpTemp = new cRtmpClient(this,
                                           socketDescriptor,
                                           this);
    tcpTemp->setmaxchunksize(this->getmaxchunksize());

    tcpClients[socketDescriptor] = tcpTemp;//插入到连接信息中

    connect(tcpTemp,&cRtmpClient::sockLog,
            this,&cRtmpServer::sockLogSlot,
            Qt::QueuedConnection);
    connect(tcpTemp,&cRtmpClient::sockDisConnect,
            this,&cRtmpServer::sockDisConnectSlot,
            Qt::QueuedConnection);
}

void cRtmpServer::sockConnectSlot(qintptr conn)
{
    showLog(QString::asprintf("%d",conn)+
            QString::fromLocal8Bit("连接上服务器."));
}

/**
 * @brief cRtmpServer::getSession 得到指定组里所有的客户端
 * @param sessionName
 * @return
 */
QList<cRtmpClient *> cRtmpServer::getSession(QString sessionName)
{
    QMutexLocker locker(&m_clientmutex);

    if(sessionName.isEmpty())
        return QList<cRtmpClient *>();

    QHash<QString, QList<cRtmpClient *>>::iterator iter = sessionClients.find(sessionName);
    if(iter == sessionClients.end())
        return QList<cRtmpClient *>();

    return iter.value();
}

void cRtmpServer::SetMetaData(AmfObjects metaData)
{
    meta_data_ = metaData;
}

void cRtmpServer::SendMediaData(QString sessionName,uint8_t type,
                                uint64_t timestamp, QByteArray data,
                                int datasize)
{
    //m_clientmutex.lock();

    if (m_gop_params[sessionName].max_gop_cache_len_ > 0) {
        this->SaveGop(sessionName,type, timestamp, data, datasize);
    }

    QList<cRtmpClient *> pClients = getSession(sessionName);
    QList<cRtmpClient *>::iterator iter = pClients.begin();
    for(;iter != pClients.end();++iter)
    {
        if((*iter)->getclientType() != PLAYER ||
            !(*iter)->isConnected())
            continue;

        if (!(*iter)->IsPlaying()) {
            (*iter)->SendMetaData(meta_data_);
            (*iter)->SendMediaData(RTMP_AVC_SEQUENCE_HEADER, 0, this->avc_sequence_header_, this->avc_sequence_header_size_);
            //(*iter).client->SendMediaData(RTMP_AAC_SEQUENCE_HEADER, 0, this->aac_sequence_header_, this->aac_sequence_header_size_);
            this->SendGop(sessionName,(*iter));
        }

        (*iter)->SendMediaData(type, timestamp, data,datasize);
    }

    //m_clientmutex.unlock();
}

void cRtmpServer::SendMetaData(QString sessionName,AmfObjects& metaData)
{
    //m_clientmutex.lock();

    QList<cRtmpClient *> pClients = getSession(sessionName);
    QList<cRtmpClient *>::iterator iter = pClients.begin();
    for(;iter != pClients.end();++iter)
    {
        if((*iter)->getclientType() != PLAYER ||
           !(*iter)->isConnected())
            continue;

        (*iter)->SendMetaData(metaData);
    }

    //m_clientmutex.unlock();
}

void cRtmpServer::SaveGop(QString sessionName,uint8_t type, uint64_t timestamp, QByteArray data, uint32_t size)
{
    uint8_t *payload = (uint8_t *)data.data();
    uint8_t frame_type = 0;
    uint8_t codec_id = 0;
    AVFrame av_frame;
    QList<AVFrame> gop;

    QHash<QString,QMap<uint64_t, QList<AVFrame>>>::iterator iter = gop_cache_.find(sessionName);
    if(iter == gop_cache_.end())
        return;

    QMap<uint64_t, QList<AVFrame>> p_gop_cache_ = iter.value();

    if (p_gop_cache_.size() > 0) {
        gop = p_gop_cache_[m_gop_params[sessionName].gop_index_];
    }

    if (type == RTMP_VIDEO) {
        frame_type = (payload[0] >> 4) & 0x0f;
        codec_id = payload[0] & 0x0f;
        if (frame_type == 1 && codec_id == RTMP_CODEC_ID_H264) {
            if (payload[1] == 1) {
                if (m_gop_params[sessionName].max_gop_cache_len_ > 0) {
                    if (p_gop_cache_.size() == 2) {
                        p_gop_cache_.erase(p_gop_cache_.begin());
                    }
                    m_gop_params[sessionName].gop_index_ += 1;
                    gop = QList<AVFrame>();
                    p_gop_cache_[m_gop_params[sessionName].gop_index_] = gop;
                    av_frame = AVFrame();
                }
            }
        }
        else if (codec_id == RTMP_CODEC_ID_H264) {
            if (m_gop_params[sessionName].max_gop_cache_len_ > 0 && gop.size() >= 1 && gop.size() < m_gop_params[sessionName].max_gop_cache_len_) {
                av_frame = AVFrame();
            }
        }
    }

    av_frame.type = type;
    av_frame.timestamp = timestamp;
    av_frame.size = size;
    av_frame.data.resize(size);
    memcpy(av_frame.data.data(), data.data(), size);
    gop.push_back(av_frame);
}

void cRtmpServer::SendGop(QString sessionName,cRtmpClient* conn)
{
    if(conn == NULL || gop_cache_.isEmpty()) return;

    QHash<QString,QMap<uint64_t, QList<AVFrame>>>::iterator iter2 = gop_cache_.find(sessionName);
    if(iter2 == gop_cache_.end() || iter2.value().isEmpty())
        return;

    QList<AVFrame> avframelist = iter2.value().begin().value();
    QList<AVFrame>::iterator iter = avframelist.begin();
    for (;iter != avframelist.end();++iter)
    {
        if (iter->type == RTMP_VIDEO) {
            conn->SendVideoData(iter->timestamp, iter->data, iter->size);
        }
    }
}

/**
 * @brief cRtmpServer::delSession 从指定组里清除指定客户端
 * @param sessionName
 * @param client
 * @return
 */
bool cRtmpServer::delSession(QString sessionName,cRtmpClient *client)
{
    if(sessionClients.isEmpty() ||
       sessionName.isEmpty() ||
       client == NULL)
        return false;

    QMutexLocker locker(&m_clientmutex);

    QHash<QString, QList<cRtmpClient *>>::iterator iter = sessionClients.find(sessionName);
    if(iter == sessionClients.end())
        return false;

    QList<cRtmpClient *>::iterator iter2 = qFind(iter.value().begin(),
                                                 iter.value().end(),
                                                 client);
    if(iter2 == iter.value().end())
        return false;

    iter.value().erase(iter2);

    if(iter.value().isEmpty())
        sessionClients.erase(iter);

    if(client->getclientType() == PUBLISHER)
    {
        gop_cache_.clear();
        m_gop_params[sessionName].gop_index_ = 0;
    }

    return true;
}

/**
 * @brief cRtmpServer::handleRtmpServerHeartTimeOut 处理心跳消息
 */
void cRtmpServer::handleRtmpServerHeartTimeOut(void)
{
    QMutexLocker locker(&m_clientmutex);

    QHash<qintptr, cRtmpClient *>::iterator iter = tcpClients.begin();
    for(;iter != tcpClients.end();++iter)
    {
        // 心跳计数
        iter.value()->HeartCount();

        // 10次心跳后，关闭这个客户端
        if(iter.value()->getHeartCount() > 10)
        {
            iter.value()->onDisconnected();
        }
    }
}

/**
 * @brief cRtmpServer::HasPublisher 检测指定名称的发布者是否存在
 * @param sessionName
 * @return
 */
bool cRtmpServer::HasPublisher(QString sessionName)
{
    if(sessionName.isEmpty() || sessionClients.isEmpty())
        return false;

    QMutexLocker locker(&m_clientmutex);

    QHash<QString, QList<cRtmpClient *>>::iterator iter = sessionClients.find(sessionName);
    if(iter == sessionClients.end())
    {
        return false;
    }

    bool isOk = false;

    QList<cRtmpClient *>::iterator iter2 = iter.value().begin();
    for(;iter2 != iter.value().end();++iter2)
    {
        if((*iter2)->getclientType() == PUBLISHER)
        {
            isOk = true;
            break;
        }
    }

    return isOk;
}

/**
 * @brief cRtmpServer::addSession 添加到一个组中
 * @param client
 */
bool cRtmpServer::addSession(QString sessionName,cRtmpClient * pclient)
{
    if(sessionName.isEmpty() || pclient == NULL)
        return false;

    QMutexLocker locker(&m_clientmutex);

    QHash<QString, QList<cRtmpClient *>>::iterator iter = sessionClients.find(sessionName);
    if(iter == sessionClients.end())
    {
        sessionClients[sessionName].push_back(pclient);
        return true;
    }

    QList<cRtmpClient *>::iterator iter2 = qFind(iter.value().begin(),
                                                 iter.value().end(),
                                                 pclient);
    if(iter2 != iter.value().end())
        return false;

    iter.value().push_back(pclient);

    if(pclient->getclientType() == PUBLISHER)
    {
        gop_cache_.clear();
        m_gop_params[sessionName].gop_index_ = 0;
    }

    return true;
}

/**
 * @brief cRtmpServer::deleteAllClients 从系统中清除所有客户端
 */
void cRtmpServer::deleteAllClients(void)
{
    QMutexLocker locker(&m_clientmutex);

    QHash<qintptr, cRtmpClient *>::iterator iter = tcpClients.begin();
    for(;iter != tcpClients.end();++iter)
    {
        iter.value()->deleteLater();
    }

    tcpClients.clear();
    sessionClients.clear();
    gop_cache_.clear();
    m_gop_params.clear();
}

/**
 * @brief cRtmpServer::deleteTheClient 从系统中清除指定客户端
 * @param conn
 */
void cRtmpServer::deleteTheClient(qintptr conn)
{
    QMutexLocker locker(&m_clientmutex);

    QHash<qintptr, cRtmpClient *>::iterator iter = tcpClients.find(conn);
    if(iter != tcpClients.end())
    {
        QHash<QString, QList<cRtmpClient *>>::iterator iter2 = sessionClients.begin();
        for(;iter2 != sessionClients.end();++iter2)
        {
            bool isFinded = false;

            QList<cRtmpClient *>::iterator iter3 = qFind(iter2.value().begin(),
                                                     iter2.value().end(),
                                                     iter.value());
            if(iter3 != iter2.value().end())
            {
                iter2.value().erase(iter3);
                isFinded = true;
            }

            if(isFinded)
            {
                if(iter2.value().isEmpty())
                    sessionClients.erase(iter2);

                break;
            }
        }

        iter.value()->deleteLater();

        tcpClients.erase(iter);
    }
}

cRtmpClient * cRtmpServer::getClient(qintptr socketDescriptor)
{
    QMutexLocker locker(&m_clientmutex);

    QHash<qintptr, cRtmpClient *>::iterator iter = tcpClients.find(socketDescriptor);
    if(iter != tcpClients.end())
    {
        return iter.value();
    }

    return NULL;
}

void cRtmpServer::sockDisConnectSlot(qintptr conn)
{
    cRtmpClient *pClient = this->getClient(conn);
    if(pClient)
    {
        showLog(QString::asprintf("[%s] %d %s:",
                                  pClient->getclientType() == PUBLISHER ? "Publish" : "Play",
                                  pClient->getcurSocketId(),
                                  pClient->getstreampath().toStdString().c_str())+
                QString::fromLocal8Bit("离开服务器."));
    }

    deleteTheClient(conn);
}

/**
 * @brief cRtmpServer::sockLogSlot 日志消息
 * @param log
 */
void cRtmpServer::sockLogSlot(QString log)
{
    showLog(log);
}

/**
 * @brief cRtmpServer::showLog
 * @param log
 */
void cRtmpServer::showLog(QString log)
{
    if(log.isEmpty()) return;

    QLOG_INFO()<<log;
    emit sockLog(log);
}

/**
 * @brief cRtmpServer::sockBinMessageSlot 处理网络二进制消息
 * @param conn
 * @param data
 */
void cRtmpServer::sockBinMessageSlot(qintptr conn,QByteArray &data)
{

}
